package com.udf.flink.demo.stream.state.checkpoint;


import com.udf.flink.demo.stream.function.BehaviorParseMapFunction;
import org.apache.flink.api.common.serialization.SimpleStringSchema;
import org.apache.flink.streaming.api.CheckpointingMode;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.CheckpointConfig;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer;

import java.util.Properties;

/**
 * Checkpoint Demo
 * Created by wy on 2020/11/22.
 */
public class CheckpointExample {
    public static void main(String[] args) throws Exception{
        Properties props = new Properties();
        props.put("bootstrap.servers", "localhost:9092");
        props.put("group.id", "metric-group");
        props.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
        props.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
        props.put("auto.offset.reset", "latest");

        String topic = "weibo_behavior";
        FlinkKafkaConsumer<String> consumer = new FlinkKafkaConsumer<>(topic, new SimpleStringSchema(), props);

        final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.enableCheckpointing(1000);

        // set mode to exactly-once (this is the default)
        env.getCheckpointConfig().setCheckpointingMode(CheckpointingMode.EXACTLY_ONCE);

        // make sure 500 ms of progress happen between checkpoints
        env.getCheckpointConfig().setMinPauseBetweenCheckpoints(500);

        // checkpoints have to complete within one minute, or are discarded
        env.getCheckpointConfig().setCheckpointTimeout(60000);

        // allow only one checkpoint to be in progress at the same time
        env.getCheckpointConfig().setMaxConcurrentCheckpoints(1);

        // enable externalized checkpoints which are retained after job cancellation
        env.getCheckpointConfig().enableExternalizedCheckpoints(CheckpointConfig.ExternalizedCheckpointCleanup.RETAIN_ON_CANCELLATION);

        // allow job recovery fallback to checkpoint when there is a more recent savepoint
//        env.getCheckpointConfig().setPreferCheckpointForRecovery(true);

        // enables the experimental unaligned checkpoints
        env.getCheckpointConfig().enableUnalignedCheckpoints();

        env.getCheckpointConfig().setTolerableCheckpointFailureNumber(1);

        DataStreamSource<String> source = env.addSource(consumer).setParallelism(1);
        source.map(new BehaviorParseMapFunction())
                .setParallelism(1)
                .uid("behavior-parse-map-function");

        env.execute("checkpoint-stream");
    }
}

/*
env.enableCheckpointing(60000)，1 分钟触发一次 checkpoint；
setCheckpointTimeout，checkpoint 超时时间，默认是 10 分钟超时，超过了超时时间就会被丢弃；
setCheckpointingMode，设置 checkpoint 语义，可以设置为 EXACTLY_ONCE，表示既不重复消费也不丢数据；AT_LEAST_ONCE，表示至少消费一次，可能会重复消费；
setMinPauseBetweenCheckpoints，两次 checkpoint 之间的间隔时间。假如设置每分钟进行一次 checkpoint，两次 checkpoint 间隔时间为 30s。假设某一次 checkpoint 耗时 40s，那么理论上20s 后就要进行一次 checkpoint，但是设置了两次 checkpoint 之间的间隔时间为 30s，所以是 30s 之后才会进行 checkpoint。另外，如果配置了该参数，那么同时进行的 checkpoint 数量只能为 1；
enableExternalizedCheckpoints，Flink 任务取消后，外部 checkpoint 信息是否被清理。
DELETE_ON_CANCELLATION，任务取消后，所有的 checkpoint 都将会被清理。只有在任务失败后，才会被保留；
RETAIN_ON_CANCELLATION，任务取消后，所有的 checkpoint 都将会被保留，需要手工清理。
setPreferCheckpointForRecovery，恢复任务时，是否从最近一个比较新的 savepoint 处恢复，默认是 false；
enableUnalignedCheckpoints，是否开启试验性的非对齐的 checkpoint，可以在反压情况下极大减少 checkpoint 的次数;
StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
// 每 1000ms 开始一次 checkpoint
env.enableCheckpointing(1000);
// 高级选项：
// 设置模式为精确一次 (这是默认值)
env.getCheckpointConfig().setCheckpointingMode(CheckpointingMode.EXACTLY_ONCE);
// 确认 checkpoints 之间的时间会进行 500 ms
env.getCheckpointConfig().setMinPauseBetweenCheckpoints(500);
// Checkpoint 必须在一分钟内完成，否则就会被抛弃
env.getCheckpointConfig().setCheckpointTimeout(60000);
// 同一时间只允许一个 checkpoint 进行
env.getCheckpointConfig().setMaxConcurrentCheckpoints(1);
// 开启在 job 中止后仍然保留的 externalized checkpoints
env.getCheckpointConfig().enableExternalizedCheckpoints(ExternalizedCheckpointCleanup.RETAIN_ON_CANCELLATION);
// 允许在有更近 savepoint 时回退到 checkpoint
env.getCheckpointConfig().setPreferCheckpointForRecovery(true);
 */